package com.intellif.mozping;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Date;
import java.util.Properties;

/**
 * @author by mozping
 * @Classname MyConsumer
 * @Description TODO
 * @Date 2019/4/16 19:35
 */
public class MyConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        //设置bootstrap.servers
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.11.63:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "false");
        props.put("auto.offset.reset", "earliest");
        props.put("max.poll.records", 300);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "90000");

        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList("test"));
        int i = 0;
        long begin;
        float total;
        int count = 10000;

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(3000);
            begin = System.currentTimeMillis();
            int num = 0;
            for (ConsumerRecord<String, String> record : records) {
                num++;
                if (record.offset() % count == 0) {
                    System.out.printf("partition = %s, offset = %d,  value = %s%n, time = %s",
                            record.partition(),
                            record.offset(),
                            record.value().substring(0, 10),
                            new Date()
                    );
                    System.out.println();
                }

            }
            consumer.commitSync();
            total = System.currentTimeMillis() - begin;
            System.out.println("消费了" + num + "条消息，耗时" + total + " ms" + new Date());
        }
    }

}
